package com.wq.concurrency.aqs;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author: wangqiang20995
 * @Date:2019/2/1
 * @Description:使用可重入锁的condition实现生产者-消费者们模式
 * @Resource:
 */
public class ProducerConsumerDemo<T> {

    private ExecutorService producers;
    private ExecutorService consumers;
    private static final Integer CORE_NUMBER = 50;
    private static final Integer RUNNABLE_WAITER = 100;
    private final Integer producerNumber;
    private final Integer consumerNumber;

    private final SafeQueue<T> safeQueue;

    public ProducerConsumerDemo(int producers,int consumers,int queue){
        this.producers = new ThreadPoolExecutor(CORE_NUMBER,100,10,
                TimeUnit.SECONDS,new ArrayBlockingQueue<>(RUNNABLE_WAITER));

        this.consumers = new ThreadPoolExecutor(CORE_NUMBER,100,10,
                TimeUnit.SECONDS,new ArrayBlockingQueue<>(RUNNABLE_WAITER));

        this.producerNumber = producers;
        this.consumerNumber = consumers;

        this.safeQueue = new SafeQueue<>(queue);

    }
    public static void main(String[] args) {
        ProducerConsumerDemo<String> consumerDemo = new ProducerConsumerDemo<>(2,1,5);
        consumerDemo.runProducer();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            System.out.println("主线程休眠异常");
        }
        consumerDemo.runConsumer();



    }

    public void runProducer() {
        for(int i = 0;i<this.producerNumber;i++){
            this.producers.execute(() -> {
                while (true) {

                    try {
                        String timeStamp = new SimpleDateFormat("yyyyMmdd HH:mm:ss").format(new Date());
                        safeQueue.put((T) timeStamp);
                        System.out.println("生产者线程[" + Thread.currentThread().getName() + "]生产一个时间[" + timeStamp + "]");

                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        System.out.println("生产者线程[" + Thread.currentThread().getName() + "]在满队列上等待空位置出现异常");
                    }
                }
            });

        }
    }

    public void runConsumer(){
        for(int i = 0;i<this.consumerNumber;i++){
            this.consumers.execute(() -> {
                while (true) {
                    try {
                        String timeStamp = (String) safeQueue.take();
                        System.out.println("消费者线程[" + Thread.currentThread().getName() + "]获取一个时间戳-->[" + timeStamp + "]");

                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        System.out.println("消费者线程[" + Thread.currentThread().getName() + "]从空队列中等待获取消息异常");
                    }
                }

            });

        }
    }

    private class Holder<T> {
        private volatile T item;

        public Holder(T t) {
            this.item = t;
        }

        public Holder() {
        }

        public void setItem(T t) {
            this.item = t;
        }

        public T getItem() {
            return this.item;
        }
    }

    private class SafeQueue<T> {
        private ReentrantLock reentrantLock;

        private Condition notEmpty;

        private Condition notFull;

        private int length, count, putPosition, takePosition;

        private final Holder<T>[] items;

        public SafeQueue(int length) {
            reentrantLock = new ReentrantLock();
            this.length = length;
            this.items = new Holder[this.length];

            this.notEmpty = reentrantLock.newCondition();
            this.notFull = reentrantLock.newCondition();


        }

        public void put(T item) throws InterruptedException {
            reentrantLock.lock();

            try {

                while (this.count == this.length) {
                    notFull.await();
                }

                this.items[this.putPosition] = new Holder<>(item);

                if (++this.putPosition == this.length) {
                    this.putPosition = 0;
                }

                this.count++;

                this.notEmpty.signal();
            } finally {
                reentrantLock.unlock();
            }
        }

        public T take() throws InterruptedException {
            reentrantLock.lock();
            try {

                while (this.count == 0) {
                    notEmpty.await();
                }

                Holder<T> holder = this.items[this.takePosition];
                if (++this.takePosition == this.length) {
                    this.takePosition = 0;
                }

                this.count--;

                this.notFull.signal();

                return holder.getItem();
            }finally {
                reentrantLock.unlock();
            }
        }
    }
}
